package com.atguigu.day02;


import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.RichParallelSourceFunction;

/**
 *
 */
public class Example10_1 {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        env
                .addSource(new RichParallelSourceFunction<String>() {
                    @Override
                    public void open(Configuration parameters) throws Exception {
                        super.open(parameters);
                        System.out.println("开始阶段：  子任务索引为:"+getRuntimeContext().getIndexOfThisSubtask());
                    }

                    @Override
                    public void close() throws Exception {
                        super.close();
                        System.out.println("结束阶段：  子任务索引为:"+getRuntimeContext().getIndexOfThisSubtask());

                    }

                    @Override
                    public void run(SourceContext<String> sourceContext) throws Exception {
                        for (int i = 0; i < 10; i++) {
                            if (i%getRuntimeContext().getNumberOfParallelSubtasks()==getRuntimeContext().getAttemptNumber()){
                                sourceContext.collect("输出*************："+i+"  索引下标："+getRuntimeContext().getIndexOfThisSubtask());
                            }else {
                                sourceContext.collect("放弃-------------："+i+"  索引下标："+getRuntimeContext().getIndexOfThisSubtask());
                            }
                        }
                    }
                    @Override
                    public void cancel() {
                    }
                })
                .setParallelism(2)
                .print()
                .setParallelism(2);
        env.execute();
    }
}
